package org.example.mq;
import com.dingtalk.api.DefaultDingTalkClient;
import com.dingtalk.api.DingTalkClient;
import com.dingtalk.api.request.OapiRobotSendRequest;
import com.dingtalk.api.response.OapiRobotSendResponse;
import com.taobao.api.ApiException;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {
        // Initialize Consumer and set Consumer Goup Name
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("F21015101");

        // Set the address of NameServer
        consumer.setNamesrvAddr("192.168.233.66:9876");
        // Subscribe One or more of topics，and specify the tag filtering conditions, here specify * means receive all tag messages
        consumer.subscribe("F21015101", "*");
        // Register a callback interface to handle messages received from the Broker
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                for (MessageExt msg: msgs) {
                    sendDingTalkMsg(new String(msg.getBody()));
                }
                // Return to the message consumption status, ConsumeConcurrentlyStatus.CONSUME_SUCCESS for successful consumption
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // Start Consumer
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
    private static void sendDingTalkMsg(String content) {
        DingTalkClient client = new DefaultDingTalkClient("https://oapi.dingtalk.com/robot/send?access_token=d47f50ebbecc488857eda4a762f7b50a7bd08f5593a67266a1b40078eeba1740");
        OapiRobotSendRequest request = new OapiRobotSendRequest();
        request.setMsgtype("text");
        OapiRobotSendRequest.Text text = new OapiRobotSendRequest.Text();
        text.setContent(content);
        request.setText(text);
        OapiRobotSendRequest.At at = new OapiRobotSendRequest.At();
        at.setIsAtAll(false);
        request.setAt(at);

        try {
            OapiRobotSendResponse response = client.execute(request);
        } catch (ApiException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

}